package com.katesoft.scale4j.rttp.jobs.tasklets;

import static com.katesoft.scale4j.common.services.IBeanNameReferences.HIBERNATE_TEMPLATE;

import java.sql.SQLException;
import java.util.Iterator;
import java.util.Map.Entry;

import net.jcip.annotations.ThreadSafe;

import org.hibernate.HibernateException;
import org.hibernate.Session;
import org.hibernate.cache.CacheKey;
import org.hibernate.cache.entry.CacheEntry;
import org.hibernate.event.EventSource;
import org.hibernate.persister.entity.AbstractEntityPersister;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.scope.context.ChunkContext;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.orm.hibernate3.HibernateCallback;

import com.hazelcast.core.IMap;
import com.katesoft.scale4j.common.lang.RuntimeUtility;
import com.katesoft.scale4j.persistent.client.LocalHibernateTemplate;
import com.katesoft.scale4j.persistent.model.unified.AbstractPersistentEntity;
import com.katesoft.scale4j.rttp.client.ClusterIsNotActiveException;
import com.katesoft.scale4j.rttp.client.IClusterInstanceAware.ClusterInstanceAwareBean;
import com.katesoft.scale4j.rttp.client.ISpringContextAccessor;

/**
 * this class will clean cache if necessary.
 * 
 * @author kate2007
 */
@ThreadSafe
public class CacheCleanTasklet extends ClusterInstanceAwareBean implements Tasklet {
   private static final long serialVersionUID = -4602303665276929268L;
   public static final String ENTITY_CLASS = "rttp.cache.clean.entity.class";

   @Override
   public RepeatStatus execute(@SuppressWarnings("unused") StepContribution contribution,
            ChunkContext chunkContext) {
      final String region = chunkContext.getStepContext().getStepExecution().getJobParameters()
               .getString(ENTITY_CLASS);
      logger.info(String.format("cache clean tasklet triggered on %s region", region));
      if (getInstance().getLifecycleService().isRunning()) {
         final IMap<Object, Object> map = getInstance().getMap(region);
         final LocalHibernateTemplate template = ISpringContextAccessor.ACCESSOR.getBean(
                  HIBERNATE_TEMPLATE, LocalHibernateTemplate.class);
         template.execute(new HibernateCallback<Object>() {
            @Override
            public Object doInHibernate(Session session) throws HibernateException, SQLException {
               for (Iterator<Entry<Object, Object>> iterator = map.entrySet().iterator(); iterator
                        .hasNext();) {
                  Entry<Object, Object> entry = iterator.next();
                  CacheEntry cacheEntry = (CacheEntry) entry.getValue();
                  CacheKey cacheKey = (CacheKey) entry.getKey();
                  AbstractPersistentEntity obj = (AbstractPersistentEntity) RuntimeUtility
                           .newInstance(region);
                  AbstractEntityPersister p = ((AbstractEntityPersister) template
                           .getSessionFactory().getClassMetadata(obj.getClass()));
                  cacheEntry.assemble(obj, cacheKey.getKey(), p, null, (EventSource) session);
                  if (obj.isCleanable()) {
                     iterator.remove();
                  }
               }
               RuntimeUtility.gc();
               return null;
            }
         });
         int oldSize = map.size();
         logger.info("region[%s] cleaned ::: old_size = %s, new_size = %s", region, oldSize,
                  map.size());
      } else {
         throw new ClusterIsNotActiveException(getInstance());
      }
      return RepeatStatus.FINISHED;
   }
}
